package org.example.work1.order;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.example.work1.order.rocketmq.ConsumerService;
import org.mybatis.spring.annotation.MapperScan;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.builder.SpringApplicationBuilder;

import java.util.List;

@SpringBootApplication
@MapperScan("org.example.work1.order")
@Slf4j
public class OrderApplication implements CommandLineRunner {

    public static void main(String[] args) {
        new SpringApplicationBuilder(OrderApplication.class).run(args);
    }


    @Override
    public void run(String... args) throws Exception {
        consumerAccount();
    }

    @Autowired
    private ConsumerService consumerService;

    public void consumerAccount() throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("accountDemo");
        consumer.setNamesrvAddr("127.0.0.1:9876");
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.subscribe("accountMessage", "*");
        /*
         *  Register callback to execute on arrival of messages fetched from brokers.
         *  注册回调函数，以便在从代理获取的消息到达时执行
         */
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> message, ConsumeConcurrentlyContext context) {
                ConsumeConcurrentlyStatus concurrentlyStatus=null;
                try {
                    concurrentlyStatus= consumerService.consumeAccountMessage(message);
                }catch (Exception e){
                    throw new RuntimeException("消费端消息消费失败",e);
                }
                return concurrentlyStatus;
            }
        });
        /*
         *  Launch the consumer instance.
         *  启动使用者实例
         */
        consumer.start();
        log.info("order模块消费account开始");
    }
}

